#include "I_Thread.h"
#include <string.h>
#include <iostream>
#include <zlib/net/I_UnixNet.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/eventfd.h>

#define MAX_THREAD_NAME_LENGTH 16
#define THREAD_MUTEX_THREAD_HOLDING (-1024 * 1024)

bool event_loop_run = true;

const int DELATY_FOR_RETRY = ink_hrtime_from_msecond(3);
const int DELATY_FOR_EMPTY = ink_hrtime_from_msecond(10);

static ink_thread_key thread_key_init();

ink_thread_key Thread::thread_data_key = thread_key_init();
ink_hrtime Thread::cur_time = 0;

static void
key_destructor(void *value)
{
	(void)value;
}

static ink_thread_key
thread_key_init()
{
	ink_thread_key_create(&Thread::thread_data_key, key_destructor);
	return Thread::thread_data_key;
}

struct thread_data_internal{
	ThreadFunction f;
	void *a;
	Thread *me;
	char name[MAX_THREAD_NAME_LENGTH];
};

static void *
spaw_thread_internal(void *a)
{
	thread_data_internal *p = (thread_data_internal *)a;

	p->me->set_specific();
	//set thread name.
	ink_set_thread_name(p->name);
	if(p->f)
		p->f(p->name);
	else
		p->me->execute();

	delete p;
	return NULL;
}

ink_thread
Thread::start(const char *name, ThreadFunction f, void *a, int detached, size_t stacksize, void *stack)
{
	thread_data_internal *p = new thread_data_internal;
	p->f = f;
	p->a = a;
	p->me = this;
	memset(p->name, 0, MAX_THREAD_NAME_LENGTH);
	if(name != NULL) {
		strncpy(p->name, name, MAX_THREAD_NAME_LENGTH-1);
		p->name[MAX_THREAD_NAME_LENGTH - 1] = '\0';
	}
	tid = ink_thread_create(spaw_thread_internal, (void *)p, detached, stacksize, stack);

	return tid;
}

void
Thread::set_specific()
{
	ink_thread_setspecific(Thread::thread_data_key, (void *)this);	
}

Thread::Thread()
{
	mutex_ptr = new_ProxyMutex();
	MUTEX_TAKE_LOCK(mutex_ptr, (EThread *)this);
	mutex_ptr.m_ptr->nthread_holding = THREAD_MUTEX_THREAD_HOLDING;
}

Thread::~Thread()
{
	//nothing
}

EThread::EThread()
	: id(-1), ethreads_tobe_signalled(NULL), ep(NULL), evfd(-1),
	n_ethreads_tobe_signalled(0), signal_hook(NULL), thread_private_data(NULL)
{

}

EThread::EThread(int anid)
	: id(anid), ethreads_tobe_signalled(NULL), ep(NULL), evfd(-1),
	n_ethreads_tobe_signalled(0), signal_hook(NULL), thread_private_data(NULL)
{
	ethreads_tobe_signalled = (EThread **)malloc(sizeof(EThread *) * MAX_EVENT_THREAD);
	memset(ethreads_tobe_signalled, 0, sizeof(EThread *) * MAX_EVENT_THREAD);
	thread_private_data = (char *)malloc(THREAD_PRIVATE_DATA);
	memset(thread_private_data, 0, sizeof(THREAD_PRIVATE_DATA));
	t = R;

	ep = new EventIO;
	evfd = eventfd(0, O_NONBLOCK | FD_CLOEXEC);
	if(evfd < 0)
	{
		assert(!"create eventfd faild!!!");
	}
}

EThread::EThread(Event *e)
	: id(-1), ethreads_tobe_signalled(NULL), ep(NULL), evfd(-1),
	n_ethreads_tobe_signalled(0), signal_hook(NULL), thread_private_data(NULL)
{
	oneevent = e;
	t = D;
}

EThread::~EThread()
{
	if(t == R)
	{
		assert(ethreads_tobe_signalled != NULL);
		assert(thread_private_data != NULL);

		free(ethreads_tobe_signalled);
		free(thread_private_data);
	}
}

void
EThread::execute()
{
	switch(t) {
	case R:
		Event *e;
		ink_hrtime next_time;

		for(;;) {
			if(!event_loop_run)
				return;
			//get current time
			cur_time = ink_get_base_hrtime_internal();
			//exec notifyQueue local queue.
			while((e = notifyQueue.dequeue_local())) {
				if(e->cancelled)
					e->free();
				else if(!e->timeout_at)
					process_event(e, e->callback_event);
				else if(e->timeout_at > 0){
					timeQueue.enqueue(e, cur_time);
				}else{ //timeout < 0
					negQueue.enqueue(e);
				}
			}

			//exec timeQueue
			bool go_on;
			do{
				go_on = false;
				timeQueue.check_ready(cur_time);
				while((e = timeQueue.dequeue())){
					if(e->cancelled){
						e->free();
						continue;
					}
					go_on = true;
					process_event(e, e->callback_event);
				}
			}while(go_on);

			//exec negative
			if(!negQueue.empty()){
				//has epoll special block
				if(n_ethreads_tobe_signalled)
					flush_signal(this);
				//exec notifyQueue -> notify local Queue
				notifyQueue.dequeue_time(cur_time, next_time, false);
				//exec notify local Queue 
            	while((e = notifyQueue.dequeue_local())) {
                	if(e->cancelled)
                    	e->free();
                	else if(!e->timeout_at)
                    	process_event(e, e->callback_event); 
                	else if(e->timeout_at > 0){
                    	timeQueue.enqueue(e, cur_time);
                	}else{ //timeout < 0
                    	negQueue.enqueue(e);
                	}
            	}
				//exec negativeQueue
				//exec poll events
				while((e = negQueue.dequeue())){
					process_event(e, EVENT_POLL);
				}
				notifyQueue.dequeue_time(cur_time, next_time, false);
			}else{
				//not has epoll special block
				next_time = timeQueue.get_near_exec_time() - cur_time;
				if(next_time > DELATY_FOR_EMPTY)
					next_time  = DELATY_FOR_EMPTY;
				if(next_time < 0)
					next_time = 0;

				if(n_ethreads_tobe_signalled)
					flush_signal(this);

				notifyQueue.dequeue_time(cur_time, next_time, true);
			}
		}

		break;
	case D:
		MUTEX_TAKE_LOCK(oneevent->mutex, this);
		oneevent->continuation->handleEvent(EVENT_IMMDIATE, oneevent);
		MUTEX_TAKE_UNLOCK(oneevent->mutex, this);
		oneevent->free();
		break;
	default:
		//bad case ThreadType
		break;
	}
}

void
EThread::process_event(Event *e, int calling_code)
{
	{
		MUTEX_TRY_LOCK(lock, e->mutex, this);
		if(!lock.is_locked()){
			e->timeout_at = DELATY_FOR_RETRY;
			notifyQueue.enqueue_local(e);
			return;
		}

		if(e->cancelled){
			assert(e->in_queue == 0);
			e->free();
			return;
		}

		e->continuation->handleEvent(calling_code, e);
	}

	if(e->period){
    // in handler maybe call event->schedule, cannot put it in queue again. (will cause double free)
    if(!e->in_queue){
		  //the PriorityQueue has exec_time
		  e->timeout_at = e->period;
		  /*if(e->period < 0)
			  e->timeout_at = e->period;
		  else{
			  cur_time = ink_get_base_hrtime_internal();
			  e->timeout_at = cur_time + e->period;
		  }*/
		  notifyQueue.enqueue_local(e);
    }
	}else{
    // in handler maybe call event->schedule, cannot free event. (will cause segment faild)
		if(!e->in_queue)
			e->free();
	}
}

Event *
EThread::schedule_imm(Continuation *cont, int callback_event)
{
	Event *e = new Event;
	e->callback_event = callback_event;
	return schedule(e->init(cont, 0, 0));	
}


Event *
EThread::schedule_imm_signal(Continuation *cont, int callback_event)
{
	Event *e = new Event;
	e->callback_event = callback_event;
	return schedule(e->init(cont, 0 ,0), true);
}


Event *
EThread::schedule_at(Continuation *cont, ink_hrtime t, int callback_event)
{
	Event *e = new Event;
	e->callback_event = callback_event;
	return schedule(e->init(cont, t, 0));
}

Event *
EThread::schedule_every(Continuation *cont, ink_hrtime t, int callback_event)
{
	Event *e = new Event;
	e->callback_event = callback_event;
	return schedule(e->init(cont, t, t));
}

Event *
EThread::schedule(Event *e, bool fast_signal)
{
	e->ethread = this;
	if(e->continuation->mutex.m_ptr)
		e->mutex = e->continuation->mutex;
	else
		e->mutex = e->continuation->mutex = e->ethread->mutex_ptr;

	notifyQueue.enqueue(e, fast_signal);
	return e;
}

Event *
EThread::schedule_imm_local(Continuation *cont, int callback_event)
{
	Event *e = new Event;
	e->callback_event = callback_event;

	return schedule_local(e->init(cont, 0, 0));
}

Event *
EThread::schedule_at_local(Continuation *cont, ink_hrtime t, int callback_event)
{
	Event *e = new Event;
	e->callback_event = callback_event;

	return schedule_local(e->init(cont, t, 0));
}

Event *
EThread::schedule_every_local(Continuation *cont, ink_hrtime t, int callback_event)
{
	Event *e = new Event;
	e->callback_event = callback_event;

	return schedule_local(e->init(cont, t, t));
}

Event *
EThread::schedule_local(Event *e)
{
	if(t != R){
		//TODO
		//eventProcessor.schedule....
		return e;
	}

	e->ethread = this;

	if(e->continuation->mutex.m_ptr){
		e->mutex = e->continuation->mutex;
	}else{
		e->mutex = e->continuation->mutex = this->mutex_ptr;
	}

	notifyQueue.enqueue_local(e);

	return e;
}
